```python
import asyncio
import os
from typing import Optional

from app.exceptions import ToolError
from app.tool.base import BaseTool, CLIResult


_BASH_DESCRIPTION = """在终端中执行一个 bash 命令。
* 长时间运行的命令：对于可能无限期运行的命令，应将其在后台运行并将输出重定向到文件，例如 command = `python3 app.py > server.log 2>&1 &`。
* 交互式：如果 bash 命令返回退出码 `-1`，这意味着进程尚未完成。助手必须发送第二次调用终端的请求，使用空 `command`（这将检索任何额外的日志），或者它可以发送额外的文本（将 `command` 设置为文本）到正在运行的进程的 STDIN，或者它可以发送 command=`ctrl+c` 来中断进程。
* 超时：如果命令执行结果显示 "Command timed out. Sending SIGINT to the process"，助手应重试在后台运行该命令。
"""


class _BashSession:
    """一个 bash shell 的会话。"""

    _started: bool  # 标记会话是否已启动
    _process: asyncio.subprocess.Process  # 当前的子进程

    command: str = "/bin/bash"  # 默认的 bash 命令路径
    _output_delay: float = 0.2  # 读取输出的延迟时间，单位为秒
    _timeout: float = 120.0  # 命令执行的超时时间，单位为秒
    _sentinel: str = "<<exit>>"  # 用于标记输出的结束

    def __init__(self):
        self._started = False  # 初始化时会话未启动
        self._timed_out = False  # 初始化时未超时

    async def start(self):
        """启动 bash 会话。"""
        if self._started:
            return  # 如果已经启动，直接返回

        # 创建一个子进程来执行 bash 命令
        self._process = await asyncio.create_subprocess_shell(
            self.command,
            preexec_fn=os.setsid,  # 设置进程组 ID
            shell=True,
            bufsize=0,
            stdin=asyncio.subprocess.PIPE,  # 标准输入管道
            stdout=asyncio.subprocess.PIPE,  # 标准输出管道
            stderr=asyncio.subprocess.PIPE,  # 标准错误管道
        )

        self._started = True  # 标记会话已启动

    def stop(self):
        """终止 bash shell。"""
        if not self._started:
            raise ToolError("会话尚未启动。")
        if self._process.returncode is not None:
            return  # 如果进程已经结束，直接返回
        self._process.terminate()  # 终止进程

    async def run(self, command: str):
        """在 bash shell 中执行一个命令。"""
        if not self._started:
            raise ToolError("会话尚未启动。")
        if self._process.returncode is not None:
            return CLIResult(
                system="工具必须重启",
                error=f"bash 已退出，返回码为 {self._process.returncode}",
            )
        if self._timed_out:
            raise ToolError(
                f"超时：bash 在 {self._timeout} 秒内未返回，必须重启",
            )

        # 我们知道这些不是 None，因为我们创建进程时使用了 PIPE
        assert self._process.stdin
        assert self._process.stdout
        assert self._process.stderr

        # 将命令发送到进程
        self._process.stdin.write(
            command.encode() + f"; echo '{self._sentinel}'\n".encode()
        )
        await self._process.stdin.drain()

        # 从进程读取输出，直到找到 sentinel
        try:
            async with asyncio.timeout(self._timeout):
                while True:
                    await asyncio.sleep(self._output_delay)
                    # 如果我们直接从 stdout/stderr 读取，它会永远等待 EOF。
                    # 使用 StreamReader 缓冲区直接读取。
                    output = (
                        self._process.stdout._buffer.decode()
                    )  # pyright: ignore[reportAttributeAccessIssue]
                    if self._sentinel in output:
                        # 去掉 sentinel 并中断
                        output = output[: output.index(self._sentinel)]
                        break
        except asyncio.TimeoutError:
            self._timed_out = True
            raise ToolError(
                f"超时：bash 在 {self._timeout} 秒内未返回，必须重启",
            ) from None

        if output.endswith("\n"):
            output = output[:-1]

        error = (
            self._process.stderr._buffer.decode()
        )  # pyright: ignore[reportAttributeAccessIssue]
        if error.endswith("\n"):
            error = error[:-1]

        # 清空缓冲区，以便可以正确读取下一次输出
        self._process.stdout._buffer.clear()  # pyright: ignore[reportAttributeAccessIssue]
        self._process.stderr._buffer.clear()  # pyright: ignore[reportAttributeAccessIssue]

        return CLIResult(output=output, error=error)


class Bash(BaseTool):
    """用于执行 bash 命令的工具"""

    name: str = "bash"  # 工具的名称
    description: str = _BASH_DESCRIPTION  # 工具的描述
    parameters: dict = {
        "type": "object",
        "properties": {
            "command": {
                "type": "string",
                "description": "要执行的 bash 命令。可以为空以在先前的退出码为 `-1` 时查看额外的日志。可以是 `ctrl+c` 来中断当前正在运行的进程。",
            },
        },
        "required": ["command"],
    }

    _session: Optional[_BashSession] = None  # 当前的 bash 会话

    async def execute(
        self, command: str | None = None, restart: bool = False, **kwargs
    ) -> CLIResult:
        """执行 bash 命令。

        Args:
            command (str | None): 要执行的命令。可以为空以在先前的退出码为 `-1` 时查看额外的日志。可以是 `ctrl+c` 来中断当前正在运行的进程。
            restart (bool): 是否重启会话。

        Returns:
            CLIResult: 命令执行结果。
        """
        if restart:
            if self._session:
                self._session.stop()  # 停止当前会话
            self._session = _BashSession()  # 创建新会话
            await self._session.start()  # 启动新会话

            return CLIResult(system="工具已重启。")

        if self._session is None:
            self._session = _BashSession()  # 创建新会话
            await self._session.start()  # 启动新会话

        if command is not None:
            return await self._session.run(command)  # 执行命令

        raise ToolError("未提供命令。")


if __name__ == "__main__":
    bash = Bash()
    rst = asyncio.run(bash.execute("ls -l"))
    print(rst)
```